package consumer

import (
	"fmt"
	"gitee.com/jianbiao/go-util/rabbitmqx/config"
	"github.com/streadway/amqp"
	"time"
)

func CreateConsumer(cnf *config.Config) (*consumer, error) {
	// 获取配置信息
	conn, err := amqp.Dial(cnf.Addr)
	queueName := cnf.QueueName
	dura := cnf.Durable
	chanNumber := cnf.ConsumerChanNumber
	reconnectInterval := time.Second * time.Duration(cnf.OffLineReconnectIntervalSec)
	retryTimes := cnf.RetryCount
	msgNum := cnf.MsgNum

	if err != nil {
		return nil, err
	}

	consumer := &consumer{
		connect:                     conn,
		queueName:                   queueName,
		durable:                     dura,
		chanNumber:                  chanNumber,
		connErr:                     conn.NotifyClose(make(chan *amqp.Error, 1)),
		offLineReconnectIntervalSec: reconnectInterval,
		retryTimes:                  retryTimes,
		msgNum: 					 msgNum,
		config:                      cnf,
	}
	return consumer, nil
}

//  定义一个消息队列结构体：WorkQueue 模型
type consumer struct {
	connect                     *amqp.Connection
	queueName                   string
	durable                     bool
	chanNumber                  int
	occurError                  error
	connErr                     chan *amqp.Error
	callbackForReceived         func(receivedData string) error //   断线重连，结构体内部使用
	offLineReconnectIntervalSec time.Duration
	retryTimes                  int
	callbackOffLine             func(err *amqp.Error) //   断线重连，结构体内部使用
	msgNum 						int //消息失败重试次数
	config                      *config.Config //配置
}

// 接收、处理消息
func (c *consumer) Received(callbackFunDealSmg func(receivedData string) error) {
	defer func() {
		_ = c.connect.Close()
	}()
	// 将回调函数地址赋值给结构体变量，用于掉线重连使用
	c.callbackForReceived = callbackFunDealSmg

	blocking := make(chan bool)

	for i := 1; i <= c.chanNumber; i++ {
		go c.receivedRun(i, callbackFunDealSmg)
	}
	<-blocking
}

func (c *consumer) receivedRun(chanNo int, callbackFunDealSmg func(receivedData string) error) {
	defer func() {
		if rerr := recover(); rerr != nil{
			fmt.Printf("rabbitmq-msg-panic:err:%v", rerr)

			go c.receivedRun(chanNo, callbackFunDealSmg)
		}
	}()
	ch, err := c.connect.Channel()
	c.occurError = errorDeal(err)
	defer func() {
		_ = ch.Close()
	}()

	q, err := ch.QueueDeclare(
		c.queueName,
		c.durable,
		!c.durable,
		false,
		false,
		nil,
	)
	c.occurError = errorDeal(err)

	err = ch.Qos(
		1,     // 大于0，服务端将会传递该数量的消息到消费者端进行待处理（通俗的说，就是消费者端积压消息的数量最大值）
		0,     // prefetch size
		false, // flase 表示本连接只针对本频道有效，true表示应用到本连接的所有频道
	)
	c.occurError = errorDeal(err)

	msgs, err := ch.Consume(
		q.Name,
		q.Name, //  消费者标记，请确保在一个消息频道唯一
		false,  //是否自动响应确认，这里设置为false，手动确认
		false,  //是否私有队列，false标识允许多个 consumer 向该队列投递消息，true 表示独占
		false,  //RabbitMQ不支持noLocal标志。
		false,  // 队列如果已经在服务器声明，设置为 true ，否则设置为 false；
		nil,
	)
	c.occurError = errorDeal(err)

	for msg := range msgs {
		//如果遇到错误，消费多次
		var msgErr error
		for msgNum := 0; msgNum< c.msgNum; msgNum++{
			// 通过回调处理消息
			msgErr = callbackFunDealSmg(string(msg.Body))
			if msgErr == nil{
				_ = msg.Ack(false) //  false 表示只确认本通道（chan）的消息
				break
			}
		}
		//超过次数，抛弃消息，记录日志
		//如果不抛弃消息的话，会再次进入队列，不断的执行那个失败的消息，会把cpu打满，并且后面的消息得不到消费，所以记录日志，等待人工处理
		if msgErr != nil{
			_ = msg.Reject(false) //手动拒绝消息 只能拒绝一条消息 为true 将再次放入队列中
			fmt.Printf("rabbitmq-msg-fail:err:%v，queueName：%v，content：%v", msgErr.Error(), c.queueName, string(msg.Body))

		}
	}
}

//消费者端，掉线重连失败后的错误回调
func (c *consumer) OnConnectionError(callbackOfflineErr func(err *amqp.Error)) {
	c.callbackOffLine = callbackOfflineErr
	go func() {
		select {
		case err := <-c.connErr:
			var i = 1
			for i = 1; i <= c.retryTimes; i++ {
				// 自动重连机制
				time.Sleep(c.offLineReconnectIntervalSec * time.Second)
				conn, err := CreateConsumer(c.config)
				if err != nil {
					continue
				} else {
					go func() {
						c.connErr = conn.connect.NotifyClose(make(chan *amqp.Error, 1))
						go conn.OnConnectionError(c.callbackOffLine)
						conn.Received(c.callbackForReceived)
					}()
					break
				}
			}
			if i > c.retryTimes {
				callbackOfflineErr(err)
			}
		}
	}()
}

// 定义一个错误处理函数
func errorDeal(err error) error {
	if err != nil {
		fmt.Printf("rabbitmq-msg-fail:err:%v", err.Error())
	}
	return err
}
